package org.example;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * @description: 生产者
 * @author: xukang
 * @create: 2025-05-29
 */
public class KafkaProducerExample {

    public static void main(String[] args) {
        // 配置生产者属性
        Properties props = new Properties();
        // ip 端口
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // key序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // value 序列化
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 我的拦截器
        // props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.example.MyInterceptor");
        // 我的分区策略
        // props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "org.example.MyPartitioner");

        // 创建生产者实例
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // 发送消息
            for (int i = 0; i < 10; i++) {
                String key = "Key-" + i;
                String value = "Value-" + i;

                ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", key, value);
                // 发送并忘记
//                producer.send(record);
                // 同步发送
//                RecordMetadata recordMetadata = producer.send(record).get();
//                System.out.println("响应结果:" + recordMetadata.toString());
                // 异步发送+回调函数
                int finalI = i;
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception != null) {
                            System.err.println("发送消息失败: " + exception.getMessage());
                        } else {
                            System.err.println("消息发送成功 - 分区: "+metadata.partition()+", 偏移量: "+metadata.offset()+",数据值:"+finalI );
                        }
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
